// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jaewon

#include "coin2/base/request/http.h"

#include <cstdlib>
#include <vector>

#include <boost/algorithm/string.hpp>
#include <boost/scope_exit.hpp>
#include <fmt/format.h>
#include <glog/logging.h>
#include <nlohmann/json.hpp>

#include "coin2/base/base64.h"

namespace coin2::base::request {

namespace {

inline bool CheckUseSsl(int port) {
  bool use_ssl = false;
  switch (port) {
    case 80:
    case 8080:
    case 8021:
    case 8086:
      use_ssl = false;
      break;
    case 443:
    case 8443:
      use_ssl = true;
      break;
    default:
      CHECK(false) << port;
      break;
  }
  return use_ssl;
}

bool GetNoDelayFromEnvVar() {
  char const* tcp_no_delay = getenv("TCP_NODELAY");
  if (tcp_no_delay != nullptr && std::string(tcp_no_delay) == "0") {
    LOG(INFO) << "Set TCP_NODELAY = false";
    return false;
  } else {
    LOG(INFO) << "Set TCP_NODELAY = true";
    return true;
  }
}

}  // namespace

HttpProxyInfo::HttpProxyInfo(const std::string& http_proxy) : http_proxy_{http_proxy} {
  ParseHttpProxy();
}

void HttpProxyInfo::ParseHttpProxy() {
  use_http_proxy_ = false;
  if (http_proxy_.empty()) {
    return;
  }
  auto tmp = http_proxy_;
  bool check_ssl_by_port = false;
  if (tmp.find(R"(https://)") != std::string::npos) {
    ssl_http_proxy_ = true;
    boost::replace_all(tmp, R"(https://)", "");
  } else if (tmp.find(R"(http://)") != std::string::npos) {
    ssl_http_proxy_ = false;
    boost::replace_all(tmp, R"(http://)", "");
  } else {
    check_ssl_by_port = true;
  }
  // retrieve proxy authrization
  if (tmp.find(R"(@)") != std::string::npos) {
    std::vector<std::string> vecs;
    boost::split(vecs, tmp, boost::is_any_of("@"));
    if (vecs.size() != 2) {
      LOG(WARNING) << "invalid http proxy, ignored. " << http_proxy_;
      return;
    }
    const auto& auth = vecs[0];
    http_proxy_auth_ = fmt::format(
        "Basic {}",
        impl::base64_encode(reinterpret_cast<const unsigned char*>(auth.c_str()), auth.size()));
    tmp = vecs[1];
  }
  // retrieve addr/port
  {
    std::vector<std::string> vecs;
    boost::split(vecs, tmp, boost::is_any_of(":"));
    if (vecs.size() != 2) {
      LOG(WARNING) << "invalid http proxy, ignored. " << http_proxy_;
      return;
    }
    http_proxy_addr_ = vecs[0];
    http_proxy_port_ = vecs[1];
  }
  if (check_ssl_by_port) {
    ssl_http_proxy_ = CheckUseSsl(stoi(http_proxy_port_));
  }
  use_http_proxy_ = true;
}

HttpsConnection::HttpsConnection(
    const std::string& host,
    int port,
    const std::string& bind_ip,
    Handler* handler,
    boost::asio::io_context* io_ctx,
    ssl::context* ssl_ctx,
    const std::string& http_proxy)
    : io_ctx_(io_ctx),
      handler_(handler),
      resolver_(*io_ctx_),
      host_(host),
      port_(port),
      bind_ip_(bind_ip),
      no_delay_(GetNoDelayFromEnvVar()),
      created_ts_(GetCurrentTimestamp()),
      http_proxy_info_{http_proxy} {
  if (http_proxy_info_.UseHttpProxy()) {
    use_ssl_ = http_proxy_info_.SslHttpProxy();
    if (CheckUseSsl(port)) {
      http_url_prefix_ = "https://";
    } else {
      http_url_prefix_ = "http://";
    }
  } else {
    use_ssl_ = CheckUseSsl(port);
  }

  if (use_ssl_) {
    stream_ssl_.reset(new beast::ssl_stream<beast::tcp_stream>(*io_ctx, *ssl_ctx));
  } else {
    stream_nossl_.reset(new beast::tcp_stream(*io_ctx));
  }

  nlohmann::json json;
  json["host"] = host_;
  json["port"] = port_;
  json["bind_ip"] = bind_ip_;
  json["no_delay"] = no_delay_;
  json["use_ssl"] = use_ssl_;
  json["http_proxy"] = http_proxy;
  LOG(INFO) << "HTTPSConnection Created." << this << ", " << json.dump();
}

HttpsConnection::HttpsConnection(
    const std::string& host,
    int port,
    Handler* handler,
    boost::asio::io_context* io_ctx,
    ssl::context* ssl_ctx,
    const std::string& http_proxy)
    : HttpsConnection(host, port, "", handler, io_ctx, ssl_ctx, http_proxy) {}

HttpsConnection::~HttpsConnection() = default;

void HttpsConnection::Start() { InitiateConnection(); }

void HttpsConnection::AsyncSend(const RequestType* req) {
  if (is_shutdown_requested()) {
    throw std::invalid_argument("shutdown requested");
  }
  if (in_sending() || !is_connected()) {
    queue_.push(req);
    return;
  }
  AsyncWrite(req);
}

void HttpsConnection::AsyncShutdown() { AsyncShutdownWithError(boost::system::error_code()); }

void HttpsConnection::AsyncShutdownWithError(boost::system::error_code ec) {
  if (shutdown_requested_) {
    return;
  }
  connected_ = false;
  shutdown_requested_ = true;

  // cancel asynchronous operations
  lowest_layer().cancel();
  // shutdown/close socket
  if (ec == boost::beast::error::timeout) {
    // https://www.boost.org/doc/libs/1_71_0/libs/beast/doc/html/beast/ref/boost__beast__basic_stream/expires_after.html
    // based on the doc, beast closed the socket if async shutdown timeout
    // OnShutdown with error code immediately
    OnShutdown(ec);
    return;
  }
  if (use_ssl_) {
    lowest_layer().expires_after(std::chrono::nanoseconds(shutdown_timeout_ns_));
    stream_ssl_->async_shutdown(
        std::bind(&HttpsConnection::OnShutdown, shared_from_this(), std::placeholders::_1));
  } else {
    // https://stackoverflow.com/questions/1993216/boostasio-cleanly-disconnecting
    boost::system::error_code ec{};
    stream_nossl_->socket().shutdown(tcp::socket::shutdown_both, ec);
  }
}

void HttpsConnection::InitiateConnection() {
  if (use_ssl_) {
    // Sets SNI Hostname. Many hosts need this to handshake successfully.
    if (!SSL_set_tlsext_host_name(stream_ssl_->native_handle(), host_.c_str())) {
      boost::system::error_code ec{
          static_cast<int>(::ERR_get_error()),
          boost::asio::error::get_ssl_category()};
      OnError("SNI", ec);
      return;
    }
  }

  auto host = host_;
  auto port = std::to_string(port_);
  if (http_proxy_info_.UseHttpProxy()) {
    host = http_proxy_info_.HttpProxyAddr();
    port = http_proxy_info_.HttpProxyPort();
  }

  // Looks up the domain name.
  resolver_.async_resolve(
      host,
      port,
      std::bind(
          &HttpsConnection::OnResolve,
          shared_from_this(),
          std::placeholders::_1,
          std::placeholders::_2));
}

void HttpsConnection::OnResolve(boost::system::error_code ec, tcp::resolver::results_type results) {
  if (is_shutdown_requested()) {
    return;
  }
  if (ec) {
    OnError("resolve", ec);
    return;
  }

  // Sets timeout.
  if (connect_timeout_ns_ > 0) {
    lowest_layer().expires_after(std::chrono::nanoseconds(connect_timeout_ns_));
  } else {
    lowest_layer().expires_never();
  }

  if (!bind_ip_.empty()) {
    lowest_layer().socket().open(boost::asio::ip::tcp::v4());
    lowest_layer().socket().bind(
        boost::asio::ip::tcp::endpoint(boost::asio::ip::address::from_string(bind_ip_), 0));

    LOG(INFO) << "LocalIP(BeforeConnect): "
              << lowest_layer().socket().local_endpoint().address().to_string() << ", " << this;

    // This is a workaround for boost::asio to use bind ip.
    lowest_layer().socket().async_connect(
        *results.begin(),  // TODO(xguo) traverse all results?
        std::bind(&HttpsConnection::OnConnect, shared_from_this(), std::placeholders::_1));
  } else {
    // Makes the connection on the IP address we get from a lookup.
    lowest_layer().async_connect(
        results,
        std::bind(&HttpsConnection::OnConnect, shared_from_this(), std::placeholders::_1));
  }
}

void HttpsConnection::OnConnect(boost::system::error_code ec) {
  if (!bind_ip_.empty()) {
    LOG(INFO) << "LocalIP(AfterConnect): "
              << lowest_layer().socket().local_endpoint().address().to_string() << ", " << this;
  }

  if (is_shutdown_requested()) {
    return;
  }
  if (ec) {
    OnError("connect", ec);
    return;
  }

  if (use_ssl_) {
    // Performs the SSL handshake.
    stream_ssl_->async_handshake(
        ssl::stream_base::client,
        std::bind(&HttpsConnection::OnHandshake, shared_from_this(), std::placeholders::_1));
  } else {
    OnHandshake(boost::system::errc::make_error_code(boost::system::errc::success));
  }
}

void HttpsConnection::OnHandshake(boost::system::error_code ec) {
  if (is_shutdown_requested()) {
    return;
  }
  if (ec) {
    OnError("handshake", ec);
    return;
  }

  connected_ = true;
  lowest_layer().expires_never();

  // set option TCP_NODELAY
  lowest_layer().socket().set_option(boost::asio::ip::tcp::no_delay(no_delay_));

  // Starts receiving responses.
  AsyncReadNext();
  ++read_count_;

  // Starts sending requests.
  if (!queue_.empty()) {
    AsyncWrite(queue_.front());
    queue_.pop();
  }

  handler_->OnConnected(shared_from_this());
}

void HttpsConnection::OnShutdown([[maybe_unused]] boost::system::error_code ec) {
  if (!ec) {
    LOG(INFO) << "HttpsConnection shutdown!: " << this << " connection age: " << age() / 1.0e9;
  } else if (ec == boost::beast::error::timeout) {
    // https://www.boost.org/doc/libs/1_71_0/libs/beast/doc/html/beast/ref/boost__beast__basic_stream/expires_after.html
    // based on the doc, beast closed the socket if async shutdown timeout
    LOG(INFO) << "HttpsConnection shutdown!: " << this << " connection age: " << age() / 1.0e9
              << " with error: " << ec.message();
    return;
  } else {
    LOG(INFO) << "HttpsConnection shutdown!: " << this << " connection age: " << age() / 1.0e9
              << " with error: " << ec.message();
  }
  if (!lowest_layer().socket().is_open()) {
    return;
  }
  io_ctx_->post([this, self = shared_from_this()]() {
    if (!lowest_layer().socket().is_open()) {
      return;
    }
    auto socket_fd = lowest_layer().socket().native_handle();
    lowest_layer().socket().close();
    LOG(INFO) << "HttpsConnection closed!: socket_fd=[" << socket_fd << "]. " << this;
  });
}

void HttpsConnection::AsyncReadNext() {
  if (use_ssl_) {
    http::async_read(
        *stream_ssl_,
        next_buffer(),
        next_response(),
        std::bind(
            &HttpsConnection::OnRead,
            shared_from_this(),
            std::placeholders::_1,
            std::placeholders::_2));
  } else {
    http::async_read(
        *stream_nossl_,
        next_buffer(),
        next_response(),
        std::bind(
            &HttpsConnection::OnRead,
            shared_from_this(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
}

void HttpsConnection::OnRead(boost::system::error_code ec, std::size_t bytes_transferred) {
  if (is_shutdown_requested()) {
    return;
  }

  BOOST_SCOPE_EXIT(&read_count_) {
    // This will be called at the end of this function.
    ++read_count_;
  }
  BOOST_SCOPE_EXIT_END;

  if (ec) {
    OnError("read", ec);
  } else {
    AsyncReadNext();
    handler_->OnResponseReceived(shared_from_this(), &current_response());
  }
}

void HttpsConnection::AsyncWrite(const RequestType* req) {
  ++request_sent_num_;
  last_request_sent_ts_ = GetCurrentTimestamp();
  in_flight_request_ = req;
  DecorateRequestForHttpProxy(req);
  if (use_ssl_) {
    http::async_write(
        *stream_ssl_,
        *req,
        std::bind(
            &HttpsConnection::OnWrite,
            shared_from_this(),
            std::placeholders::_1,
            std::placeholders::_2));
  } else {
    http::async_write(
        *stream_nossl_,
        *req,
        std::bind(
            &HttpsConnection::OnWrite,
            shared_from_this(),
            std::placeholders::_1,
            std::placeholders::_2));
  }
}

void HttpsConnection::OnWrite(
    boost::system::error_code ec,
    [[maybe_unused]] std::size_t bytes_transferred) {
  const auto* sent_requeset = in_flight_request_;
  in_flight_request_ = nullptr;
  if (is_shutdown_requested()) {
    return;
  }
  if (ec) {
    OnError("write", ec);
    return;
  }

  if (!queue_.empty()) {
    AsyncWrite(queue_.front());
    queue_.pop();
  }

  handler_->OnRequestSent(shared_from_this(), sent_requeset);
}

void HttpsConnection::OnError(const char* where, boost::system::error_code ec) {
  if (is_shutdown_requested()) {
    return;
  }
  int64_t request_sent_interval = (GetCurrentTimestamp() - last_request_sent_ts_) / 1.0e9;
  LOG(ERROR) << where << ": " << ec.message() << " connection age: " << age() / 1.0e9
             << ", request_sent_num: " << request_sent_num_
             << ", interval: " << request_sent_interval << " " << this;
  AsyncShutdownWithError(ec);
  handler_->OnError(shared_from_this(), ec);
}

void HttpsConnection::DecorateRequestForHttpProxy(const RequestType* req) {
  if (!http_proxy_info_.UseHttpProxy()) {
    return;
  }
  std::string host = port_ == 0 ? host_ : fmt::format("{}:{}", host_, port_);
  std::string target = fmt::format(R"({}{}{})", http_url_prefix_, host, std::string{req->target()});
  const_cast<RequestType*>(req)->target(target);
  const_cast<RequestType*>(req)->set(
      http::field::proxy_authorization,
      http_proxy_info_.HttpProxyAuth());
}

}  // namespace coin2::base::request
